package com.asiainfo.dacp.datastash.broker.workerctrl.service;

import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import org.apache.curator.shaded.com.google.common.collect.Maps;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.asiainfo.dacp.datastash.broker.core.BrokerModel;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitTask;
import com.asiainfo.dacp.datastash.broker.facade.UnitProcessFacade;
import com.asiainfo.dacp.datastash.broker.facade.UnitTaskFacade;
import com.asiainfo.dacp.datastash.broker.listener.AbstractQueueListener;
import com.asiainfo.dacp.datastash.broker.util.CallableResult;

@Service
public class ResultCheckService {
	@Autowired
	private UnitTaskFacade unitTaskFacade;
	@Autowired
	private UnitProcessFacade unitProcessFacade;
	@Autowired
	private BrokerModel brokerModel;

	private Map<StashUnitTask, Future<CallableResult>> running = Maps.newConcurrentMap();

	public synchronized void add(StashUnitTask task, Future<CallableResult> f) {
		running.put(task, f);
	}

	public void start() {

		new Thread(new AbstractQueueListener() {
			@Override
			protected void handle() {
				try {
					for (StashUnitTask stashUnitTask : running.keySet()) {
						Future<CallableResult> future = running.get(stashUnitTask);

						if (future.isDone()) {
							CallableResult result;
							try {
								result = future.get();
								if (result.isError()) {
									// TODO 后续考虑失败重试 ，需表结构调整 task_state 加上
									// retry_cnt 字段
									unitTaskFacade.taskFailed(stashUnitTask);
									// 失败
								} else {
									unitTaskFacade.taskComplete(stashUnitTask);
								}
								unitProcessFacade.checkUnitResult(stashUnitTask);
								brokerModel.deleteBrokerUnitTaskList(stashUnitTask.getTaskProcId());
							} catch (InterruptedException e) {
								e.printStackTrace();
							} catch (ExecutionException e) {
								e.printStackTrace();
							}
						}
					}
				} catch (Exception e) {
					e.printStackTrace();
				}
			}

			@Override
			protected int getIntervals() {
				return 5;
			}
		}).start();

	}
}
